package com.taijihuabao.common.rabbitmq;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import com.taijihuabao.common.emailutils.EmailPojo;
import com.taijihuabao.common.emailutils.EmailUtils;
import com.taijihuabao.common.exception.ServiceException;
import com.taijihuabao.common.jsonresult.ApplicationResponseCode;
import com.taijihuabao.modules.entity.ResendMessageEntity;
import com.taijihuabao.modules.mapper.ResendMessageMapper;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import javax.annotation.Resource;
import java.io.IOException;
import java.time.Duration;
import java.util.*;

/**
 * @author changhui.tu
 * MQ队列消费
 * RabbitListener:监听指定队列
 */
@Component
public class RabbitMqConsumer {

    /**
     * 常量提取
     */
    private static final String MQ_MESSAGE_AFFIRM_PREFIX = "MessageAffirm-";
    private static final String RETRY_COUNT_PREFIX = "RetrySendCount-";
    private static final String DEATH_MESSAGE = "x-death";

    @Autowired
    private ObjectMapper objectMapper;
    @Resource
    private EmailUtils emailUtils;
    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    @Resource
    private ResendMessageMapper resendMessageMapper;

    /**
     * 邮件队列消费
     *
     * @param emailBytes
     * @throws IOException
     */
    @RabbitListener(queues = RabbitmqConfig.QUEUE_MAIL_SEND)
    public void mailSendConsume(@Payload byte[] emailBytes, Channel channel, Message message) throws IOException {
        EmailPojo emailPojo = objectMapper.readValue(emailBytes, EmailPojo.class);
        //获取消息唯一id去redis查询
        String messageId = message.getMessageProperties().getMessageId();
        //是否重复消费
        if (!ObjectUtils.isEmpty(redisTemplate.opsForValue().get(MQ_MESSAGE_AFFIRM_PREFIX + messageId))) {
            return;
        }
        try {
            //throw new ServiceException(ApplicationResponseCode.ID_NOTNULL);
            emailUtils.sendSimpleTextMailActual(emailPojo);
            //成功执行手动确认，multiple表示ack所有小于getDeliveryTag(消息的index)的消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            redisTemplate.opsForValue().set(MQ_MESSAGE_AFFIRM_PREFIX + messageId, messageId, Duration.ofMinutes(RabbitMqProducer.OUT_TIME));
        } catch (Exception e) {
            int maxRetry = 4;
            int retryCount;
            Object count = redisTemplate.opsForValue().get(RETRY_COUNT_PREFIX + messageId);
            if (!ObjectUtils.isEmpty(count)) {
                retryCount = (int) count;
            } else {
                retryCount = 0;
            }
            //是否达到最大重试次数
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, retryCount < maxRetry);
            redisTemplate.opsForValue().set(RETRY_COUNT_PREFIX + messageId, retryCount + 1, Duration.ofMinutes(RabbitMqProducer.OUT_TIME));
        }
    }

    /**
     * 死信队列消费
     *
     * @param emailBytes
     * @param channel
     * @param message
     * @throws IOException
     */
    @RabbitListener(queues = RabbitmqConfig.QUEUE_DEAD_LETTER)
    public void deadLetterConsumer(@Payload byte[] emailBytes, Channel channel, Message message) throws IOException {
        String messageId = message.getMessageProperties().getMessageId();
        Map<String, Object> headers = message.getMessageProperties().getHeaders();
        if (headers != null) {
            if (headers.containsKey(DEATH_MESSAGE)) {
                List<Map<String, Object>> deaths = (List<Map<String, Object>>) headers.get("x-death");
                if (deaths.size() > 0) {
                    Map<String, Object> death = deaths.get(0);
                    ResendMessageEntity resendMessage = new ResendMessageEntity();
                    resendMessage.setMessageId(messageId);
                    resendMessage.setMessageByteArray(emailBytes);
                    resendMessage.setMessageReason((String) death.get("reason"));
                    resendMessage.setDeadTime((Date) death.get("time"));
                    resendMessage.setDeadCount((Long) death.get("count"));
                    resendMessage.setFormerExchange((String) death.get("exchange"));
                    resendMessage.setFormerQueue((String) death.get("queue"));
                    List<String> routingKeys = (List<String>) death.get("routing-keys");
                    resendMessage.setFormerRoutingKey(routingKeys.get(0));
                    if (resendMessageMapper.resendMessage(resendMessage)) {
                        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                    }
                }
            }
        }
    }

    /**
     * 计算重试次数(死信队列)
     *
     * @param properties
     * @return
     */
    private Long getRetryCount(MessageProperties properties) {
        Long retryCount = 0L;
        Map<String, Object> headers = properties.getHeaders();
        if (headers != null) {
            if (headers.containsKey(DEATH_MESSAGE)) {
                List<Map<String, Object>> deaths = (List<Map<String, Object>>) headers.get("x-death");
                if (deaths.size() > 0) {
                    Map<String, Object> death = deaths.get(0);
                    retryCount = (Long) death.get("count");
                    System.out.println("当前重试次数：" + retryCount);
                }
            }
        }
        return retryCount;
    }

}
